package com.Hsu.mq.mqserver.datacenter;

/**
 * @author Hsu琛君珩
 * @ClassName:MessageFileManager
 * @date 2024-02-14
 * @apiNote
 * @Version: v1.0
 */

import com.Hsu.mq.common.BinaryTool;
import com.Hsu.mq.common.MqException;
import com.Hsu.mq.mqserver.core.MSGQueue;
import com.Hsu.mq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.List;
import java.util.Scanner;

/**
 * 负责操作和管理硬盘上的消息文件，包括消息的读写、文件的创建和删除等
 * 通过这个类来统计硬盘上的消息进行管理
 * 这个类的设计目的是为了在消息队列系统中实现消息的持久化存储，确保即使在系统重启后消息也不会丢失
 */
public class MessageFileManager {
    //定义个内部类表示该队列的统计信息
    //一般定义内部类考虑使用 static，即 静态内部类，解耦合了一些
    static public class Stat {
        //就不设置为 private 了，没必要写个 getter setter
        public int totalCount;//总消息数量
        public int validCount;//有效消息数量
    }

    public void init(){
        //暂时不需要做什么初始化的工作，方便后续扩展
    }

    //约定消息文件所在的目录和文件名
    //用来获取指定队列对应的消息文件的所在路径
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

    //用来获取该队列的消息数据文件路径
    //注意：二进制文件一般用 .dat、.bin 表示，用 .txt 作为后缀不太合适，这一般表示文本，但我们也不改了
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    //用来获取该队列的消息统计文件路径
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    //读取队列的统计信息
    private Stat readStat(String queueName) {
        //当前的消息统计文件是文本文件，直接用 Scanner 来读取文件内容
        Stat stat = new Stat();
        try (InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
            Scanner sc=new Scanner(inputStream);
            stat.totalCount=sc.nextInt();//总消息
            stat.validCount=sc.nextInt();//有效消息
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    //写入队列的统计信息
    private void writeStat(String queueName, Stat stat) {
        //使用 PrintWriter 来写文件
        //outputStream 打开文件，默认情况下会把源文件清空，此时相当于新的数据覆盖了旧的数据
        try (OutputStream outputStream=new FileOutputStream(getQueueStatPath(queueName))){
            PrintWriter printWriter=new PrintWriter(outputStream);
            printWriter.write(stat.totalCount+"\t"+stat.validCount);
            printWriter.flush();//保证数据从缓冲区刷到硬盘中
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //创建队列对应的文件和目录
    public void createQueueFiles(String queueName) throws IOException {
        //1.先创建队列的对应的消息目录
        File baseDir=new File(getQueueDir(queueName));
        if(!baseDir.exists()){
            //不存在就创建和这个目录
            boolean ok=baseDir.mkdirs();
            if(!ok){
                throw new IOException("创建目录失败！baseDir="+baseDir.getAbsolutePath());
            }
        }
        //2.创建队列数据文件
        File queueDataFile=new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()){
            boolean ok=queueDataFile.createNewFile();
            if(!ok){
                throw new IOException("创建文件失败！queueDataFile="+queueDataFile.getAbsolutePath());
            }
        }
        //3.创建队列统计文件
        File queueStatFile=new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()){
            boolean ok=queueStatFile.createNewFile();
            if(!ok){
                throw new IOException("创建文件失败！queueStatFile="+queueStatFile.getAbsolutePath());
            }
        }
        //4.给消息统计文件，设定初始值：0\t0
        Stat stat=new Stat();
        stat.totalCount=0;
        stat.validCount=0;
        writeStat(queueName,stat);
    }

    //删除队列对应的文件和目录
    //队列被删除后，对应的消息文件等等，也会被删除
    public void destroyQueueFiles(String queueName) throws IOException {
        //先删除里面的文件，再删除目录（因为删目录时要保证目录为空）
        File queueDataFile=new File(getQueueDataPath(queueName));
        boolean ok1=queueDataFile.delete();
        File queueStatFile=new File(getQueueStatPath(queueName));
        boolean ok2=queueStatFile.delete();
        File baseDir=new File(getQueueDir(queueName));
        boolean ok3=baseDir.delete();
        if(!ok1||!ok2||!ok3){
            //有一个删除失败都算整体失败
            throw new IOException("删除队列目录和文件失败！baseDir="+baseDir.getAbsolutePath());
        }
    }

    //检查队列的目录和文件是否存在
    //后续生产者给 BrokerServer 生产消息了，这个消息可能需要记录到文件上（取决于消息是否要持久化）
    public boolean checkFileExits(String queueName){
        //判断队列的数据文件和统计文件是否都存在
        File queueDataFile=new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()){
            return false;
        }
        File queueStatFile=new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()){
            return false;
        }
        return true;
    }

    /**
     * 用来把一个新的消息放到队列对应的文件中
     * @param queue 要把消息写入的队列
     * @param message 要写的消息
     */
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        //1.检查一下当前要写入的对接对应的文件是否存在
        if(!checkFileExits(queue.getName())){
            throw new MqException("[MessageFileManager] 队列对应的文件不存在！queueName="+queue.getName());
        }
        //2.把 Message 对象进行序列化，转成二进制的字节数组
        byte[] messageBinary=BinaryTool.toBytes(message);
        synchronized (queue){//加锁，这个queue对象后续是通过内存管理的
            //3.先获取到当前的队列数据文件的长度，计算出该 Message 的 offsetBeg 和 offsetEnd，这两个就是写文件时赋值的
            //把新的 Message 数据写入到队列数据文件的末尾，offsetBeg 就是当前文件长度 + 4，offsetEnd 就是当前文件长度 + 4 + message 自身长度
            File queueDataFile=new File(getQueueDataPath(queue.getName()));
            //queueDataFile.length();//文件长度，单位字节
            message.setOffsetBeg(queueDataFile.length()+4);
            message.setOffsetEnd(queueDataFile.length()+4+messageBinary.length);
            //4.写入到消息到数据文件，这里不是覆盖写，而是追加写到数据文件末尾
            try(OutputStream outputStream=new FileOutputStream(queueDataFile,true)){//true表示追加写到末尾
                try(DataOutputStream dataOutputStream=new DataOutputStream(outputStream)) {
                    //先写消息的长度，占据 4 个字节，需要把 长度 这个int类型数据的4个字节依次写到文件中
                    //在流对象中经常用 int 表示 byte 的情况，直接用 outputStream.write(messageBinary.length) 写入就只有1个字节
                    dataOutputStream.writeInt(messageBinary.length);
                    //写入消息本体
                    dataOutputStream.write(messageBinary);
                }
            }
            //5.更新消息统计文件
            Stat stat=readStat(queue.getName());
            stat.totalCount+=1;
            stat.validCount+=1;
            writeStat(queue.getName(),stat);
        }
    }

    /**
     * 这个是删除消息的方法，这里是逻辑删除
     * 也就是把硬盘里存储的这个数据里的 isValid 属性设置为 0
     * 这里的删除就是同时从 内存和硬盘 中删除
     * @param queue
     * @param message 必须包含有效的 offsetBeg offsetEnd，这个 message 是内存管理的消息对象
     */
    public void deleteMessage(MSGQueue queue,Message message) throws IOException, ClassNotFoundException {
        synchronized (queue){//加锁
            //之前用 FileInputStream/FileInputStream 是从文件头开始读的，现在要在文件指定位置读写，随机访问
            try(RandomAccessFile randomAccessFile=new RandomAccessFile(getQueueDataPath(queue.getName()),"rw")){//可读可写
                //1.先把文件中这段数据读取出来
                byte[] bufferSrc=new byte[(int) (message.getOffsetEnd()-message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());//这个方法相当于我们平时打字时显示的移动光标，不是鼠标光标，别搞错
                randomAccessFile.read(bufferSrc);
                //2.把读取的数据，还原回 Message 对象，这个 diskMessage 是硬盘管理的消息对象
                Message diskMessage= (Message) BinaryTool.fromBytes(bufferSrc);
                //3.把 isValid 设置为 0，即无效
                diskMessage.setIsValid((byte) 0x0);//isValid 只是在文件中标识这个消息是否有效
                //此处不用给参数的 message 的 isValid 设为0，因为这个参数是内存管理的，而且很快就要从内存中销毁了
                //4.把上述数据重新写入文件
                byte[] bufferDest=BinaryTool.toBytes(diskMessage);
                //虽然上面seek过了，但之后又进行了读操作，这一读就导致文件光标往后移动，移动到下一个消息的位置了
                //要想让接下来的写入能刚好回到之前的位置，就要重新调整文件光标
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(bufferDest);
                //经过上述操作，对于文件来说只有一个字节发生改变，也就是 有效标志位
            }
            //5.更新消息统计文件，把一个消息设置为无效了，有效消息个数就-1
            Stat stat=readStat(queue.getName());
            if(stat.validCount>0){
                stat.validCount-=1;
            }
            writeStat(queue.getName(),stat);
        }
    }

    /**
     * 用这个方法从文件中读取所有的消息内容，加载到内存中，具体是放到一个链表
     * 在程序启动时进行调用
     * 用链表是因为后续方便进行头删操作
     * 这方法参数只有一个 queueName 而不是 MSGQueue，因为不用加锁，因为只需要在程序启动时调用，服务器还不能处理请求，不涉及多线程调用
     * @param queueName
     * @return
     */
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages=new LinkedList<>();
        try(InputStream inputStream=new FileInputStream(getQueueDataPath(queueName))){
            try(DataInputStream dataInputStream=new DataInputStream(inputStream)){//上面是 output 写，这里就是 input 读
                //用这个变量记录当前文件光标，也就是读到哪个位置了
                long currentOffset=0;
                //一个文件中包含了很多消息，要循环读取
                while(true){
                    //1.先读取当前消息的长度，这里的 readInt 可能会读到文件末尾(EOF)
                    //这里的 readInt 会抛出 EOFException 异常，这个和很多流对象不太一样
                    int messageSize=dataInputStream.readInt();//读4个字节
                    //2.按照这个长度读取消息内容
                    byte[] buffer=new byte[messageSize];
                    int actualSize=dataInputStream.read(buffer);
                    if(messageSize!=actualSize){
                        //如果不匹配说明文件有问题
                        throw new MqException("[MessageFileManager] 文件格式错误！queueName="+queueName);
                    }
                    //3.把读到的二进制数据反序列化回 Message 对象
                    Message message = (Message) BinaryTool.fromBytes(buffer);
                    //4.判定一下这个消息对象，是否为无效对象
                    if(message.getIsValid()!=0x1){
                        //无效数据，直接跳过，但也要计算长度
                        currentOffset+=(4+messageSize);
                        continue;
                    }
                    //5.有效数据则把这个 Message 对象加入到链表中，加入前填写 offsetBeg offsetEnd
                    //计算前需要知道当前文件光标的位置，但由于使用的 DataInputStream 并不方便直接获取到文件光标
                    //因此我们手动计算
                    message.setOffsetBeg(currentOffset+4);
                    message.setOffsetEnd(currentOffset+4+messageSize);
                    currentOffset+=(4+messageSize);//4是存储消息长度的空间，Size是消息体的长度本身
                    messages.add(message);//添加到链表
                }
            }catch (EOFException e){
                //这个 catch 不是为了处理异常，而是为了正常处理业务逻辑，文件读到末尾就被 readInt 抛出异常
                //这个 catch 语句也不用干什么，捕获到那个异常就是读到末尾了就该结束了
                System.out.println("[MessageFileManager] 恢复 Message 数据完成！");
            }
        }
        return messages;
    }

    /**
     * 检查当前是否要针对该队列的消息数据文件进行 GC
     * @param queueName
     * @return
     */
    public boolean checkGC(String queueName){
        //判定是否要 GC，是根据总消息数和有效消息数，这两个都在 消息统计文件中
        Stat stat=readStat(queueName);
        if(stat.totalCount>2000&&(double)stat.validCount/(double)stat.totalCount<0.5){
            return true;
        }
        return false;
    }

    //获取一个新消息文件，把有消息往这个文件写，最后再把这个文件命名为原来的名字 queue_data.txt
    public String getQueueDataNewPath(String queueName){
        return getQueueDir(queueName)+"/queue_data_new.txt";
    }

    /**
     * 通过这个方法真正执行消息数据文件的垃圾回收操作
     * 使用复制算法完成
     * 创建一个新的文件，名为 queue_data_new.txt
     * 把之前的消息数据文件都读取出来，写到这个新的文件
     * 删除旧文件，再把新文件改名为 queue_data.txt
     * 同时记得要更新消息统计文件
     * @param queue
     */
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        //进行 gc 时，是针对消息数据文件进行大洗牌，此时其他线程不能针对该队列的消息文件做任何修改
        synchronized (queue){//加锁
            //由于 gc 可能操作很耗时，统计一下执行消耗时间
            long gcBeg=System.currentTimeMillis();

            //1.创建一个新文件
            File queueDataNewFile=new File(getQueueDataNewPath(queue.getName()));
            if(queueDataNewFile.exists()){
                //正常情况下这个文件不应该存在，如果存在说明上次 gc 了一半程序意外崩溃了
                throw new MqException("[MessageFileManager] gc 时发现该队列的 queue_data_new 已经存在！queueName="+queue.getName());
            }
            boolean ok=queueDataNewFile.createNewFile();
            if(!ok){
                throw new MqException("[MessageFileManager] 创建文件失败！queueDataNewFile="+queueDataNewFile.getAbsolutePath());
            }
            //2.从旧文件中读取出所有的有效消息对象
            LinkedList<Message> messages=loadAllMessageFromQueue(queue.getName());
            //3.把有效消息写入新文件中
            try(OutputStream outputStream=new FileOutputStream(queueDataNewFile)){
                try(DataOutputStream dataOutputStream=new DataOutputStream(outputStream)){
                    for(Message message:messages){
                        byte[] buffer=BinaryTool.toBytes(message);
                        //先写4个字节的消息长度
                        dataOutputStream.writeInt(buffer.length);
                        //再写内容
                        dataOutputStream.write(buffer);
                    }
                }
            }
            //4.删除旧的数据文件，并且把新的文件重命名
            File queueDataOldFile=new File(getQueueDataPath(queue.getName()));
            ok=queueDataOldFile.delete();
            if(!ok){
                throw new MqException("[MessageFileManager] 删除旧的数据文件失败！queueDataOldFile="+queueDataOldFile.getAbsolutePath());
            }
            //把 queue_data_new.txt => queue_data.txt
            ok=queueDataNewFile.renameTo(queueDataOldFile);
            if(!ok){
                throw new MqException("[MessageFileManager] 文件重命名失败！queueDataNewFile="+queueDataNewFile.getAbsolutePath()+", queueDataOldFile"+queueDataOldFile.getAbsolutePath());
            }
            //5.更新消息统计文件
            Stat stat=readStat(queue.getName());
            stat.totalCount=messages.size();
            stat.validCount=messages.size();
            writeStat(queue.getName(),stat);

            long gcEnd=System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc 执行完毕！queueName="+ queue.getName()+", time="+(gcEnd-gcBeg)+"ms");
        }
    }
}